xplly'Blog

一、概述

Canal [kə'næl]，译意为水道/管道/沟渠，canal是阿里巴巴旗下的一款开源项目，基于Java开发。基于数据库增量日志解析，提供增量数据订阅&消费。GitHub的地址：https://github.com/alibaba/canal

Canal是基于mysql的主从同步来实现的，MySQL主从同步的原理如下：

1）MySQL master 将数据变更写入二进制日志( binary log），其中记录的数据叫做binary log events
2）MySQL slave 将 master 的 binary log events拷贝到它的中继日志(relay log)
3）MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

而Canal就是把自己伪装成MySQL的一个slave节点，从而监听master的binary log变化。再把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。

Canal提供了各种语言的客户端，当Canal监听到binlog变化时，会通知Canal的客户端。客户端收到变化消息时，可以对数据变化进行一些业务逻辑操作。

1.1 配置文件

canal.properties (系统根配置文件)

参数名字	参数说明	默认值
canal.id	每个canal server实例的唯一标识，暂无实际意义	1
canal.ip	canal server绑定的本地IP信息，如果不配置，默认选择一个本机IP进行启动服务	无
canal.port	canal server提供socket服务的端口	11111
canal.zkServers	canal server链接zookeeper集群的链接信息例子：10.20.144.22:2181,10.20.144.51:2181	无
canal.zookeeper.flush.period	canal持久化数据到zookeeper上的更新频率，单位毫秒	1000
canal.instance.memory.batch.mode	canal内存store中数据缓存模式 1. ITEMSIZE : 根据buffer.size进行限制，只限制记录的数量 2. MEMSIZE : 根据buffer.size * buffer.memunit的大小，限制缓存记录的大小	MEMSIZE
canal.instance.memory.buffer.size	canal内存store中可缓存buffer记录数，需要为2的指数	16384
canal.instance.memory.buffer.memunit	内存记录的单位大小，默认1KB，和buffer.size组合决定最终的内存使用大小	1024
canal.instance.transactionn.size	最大事务完整解析的长度支持超过该长度后，一个事务可能会被拆分成多次提交到canal store中，无法保证事务的完整可见性	1024
canal.instance.fallbackIntervalInSeconds	canal发生mysql切换时，在新的mysql库上查找binlog时需要往前查找的时间，单位秒说明：mysql主备库可能存在解析延迟或者时钟不统一，需要回退一段时间，保证数据不丢	60
canal.instance.detecting.enable	是否开启心跳检查	false
canal.instance.detecting.sql	心跳检查sql	insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.interval.time	心跳检查频率，单位秒	3
canal.instance.detecting.retry.threshold	心跳检查失败重试次数	3
canal.instance.detecting.heartbeatHaEnable	心跳检查失败后，是否开启自动mysql自动切换说明：比如心跳检查失败超过阀值后，如果该配置为true，canal就会自动链到mysql备库获取binlog数据	false
canal.instance.network.receiveBufferSize	网络链接参数，SocketOptions.SO_RCVBUF	16384
canal.instance.network.sendBufferSize	网络链接参数，SocketOptions.SO_SNDBUF	16384
canal.instance.network.soTimeout	网络链接参数，SocketOptions.SO_TIMEOUT	30
canal.instance.filter.druid.ddl	是否使用druid处理所有的ddl解析来获取库和表名	true
canal.instance.filter.query.dcl	是否忽略dcl语句	false
canal.instance.filter.query.dml	是否忽略dml语句 (mysql5.6之后，在row模式下每条DML语句也会记录SQL到binlog中,可参考MySQL文档)	false
canal.instance.filter.query.ddl	是否忽略ddl语句	false
canal.instance.filter.table.error	是否忽略binlog表结构获取失败的异常(主要解决回溯binlog时,对应表已被删除或者表结构和binlog不一致的情况)	false
canal.instance.filter.rows	是否dml的数据变更事件(主要针对用户只订阅ddl/dcl的操作)	false
canal.instance.filter.transaction.entry	是否忽略事务头和尾,比如针对写入kakfa的消息时，不需要写入TransactionBegin/Transactionend事件	false
canal.instance.binlog.format	支持的binlog format格式列表 (otter会有支持format格式限制)	ROW,STATEMENT,MIXED
canal.instance.binlog.image	支持的binlog image格式列表 (otter会有支持format格式限制)	FULL,MINIMAL,NOBLOB
canal.instance.get.ddl.isolation	ddl语句是否单独一个batch返回(比如下游dml/ddl如果做batch内无序并发处理,会导致结构不一致)	false
canal.instance.parser.parallel	是否开启binlog并行解析模式(串行解析资源占用少,但性能有瓶颈, 并行解析可以提升近2.5倍+)	true
canal.instance.parser.parallelBufferSize	binlog并行解析的异步ringbuffer队列 (必须为2的指数)	256
canal.instance.tsdb.enable	是否开启tablemeta的tsdb能力	true
canal.instance.tsdb.dir	主要针对h2-tsdb.xml时对应h2文件的存放目录,默认为conf/xx/h2.mv.db	$c a n a l . f i l e . d a t a . d i r : . . / c o n f /$
canal.instance.tsdb.url	jdbc url的配置(h2的地址为默认值，如果是mysql需要自行定义)	jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername	jdbc url的配置(h2的地址为默认值，如果是mysql需要自行定义)	canal
canal.instance.tsdb.dbPassword	jdbc url的配置(h2的地址为默认值，如果是mysql需要自行定义)	canal
canal.instance.rds.accesskey	aliyun账号的ak信息 (如果不需要在本地binlog超过18小时被清理后自动下载oss上的binlog，可以忽略该值)	无
canal.instance.rds.secretkey	aliyun账号的sk信息(如果不需要在本地binlog超过18小时被清理后自动下载oss上的binlog，可以忽略该值)	无

instance.properties (instance级别的配置文件，每个instance一份)

参数名字	参数说明	默认值
canal.instance.mysql.slaveId	mysql集群配置中的serverId概念，需要保证和当前mysql集群中id唯一 (v1.1.x版本之后canal会自动生成，不需要手工指定)	无
canal.instance.master.address	mysql主库链接地址	127.0.0.1:3306
canal.instance.master.journal.name	mysql主库链接时起始的binlog文件	无
canal.instance.master.position	mysql主库链接时起始的binlog偏移量	无
canal.instance.master.timestamp	mysql主库链接时起始的binlog的时间戳	无
canal.instance.gtidon	是否启用mysql gtid的订阅模式	false
canal.instance.master.gtid	mysql主库链接时对应的gtid位点	无
canal.instance.dbUsername	mysql数据库帐号	canal
canal.instance.dbPassword	mysql数据库密码	canal
canal.instance.defaultDatabaseName	mysql链接时默认schema
canal.instance.connectionCharset	mysql 数据解析编码	UTF-8
canal.instance.filter.regex	mysql 数据解析关注的表，Perl正则表达式.多个正则之间以逗号(,)分隔，转义符需要双斜杠(\) 常见例子：1. 所有表：.* or .\.. 2. canal schema下所有表： canal\..* 3. canal下的以canal打头的表：canal\.canal.* 4. canal schema下的一张表：canal.test15. 多个规则组合使用：canal\..*,mysql.test1,mysql.test2 (逗号分隔)	.\..
canal.instance.filter.black.regex	mysql 数据解析表的黑名单，表达式规则见白名单的规则	无
canal.instance.rds.instanceId	aliyun rds对应的实例id信息(如果不需要在本地binlog超过18小时被清理后自动下载oss上的binlog，可以忽略该值)	无

二、使用Demo

2.1 与springboot整合

依赖

xml

<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

配置

yml

canal:
  destination: heima # canal的集群名字，要与安装canal时设置的名称一致
  server: 192.168.150.101:11111 # canal服务地址

监听器类

java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

@CanalTable("Demo")
@Component
public class DemoHandler implements EntryHandler<Item> {

    // 监听数据插入的逻辑
    @Override
    public void insert(Item item) {
    }

    // 监听数据更新的逻辑
    @Override
    public void update(Item before, Item after) {
    }

    // 监听数据删除的逻辑
    @Override
    public void delete(Item item) {
    }
}

2.2 基础应用

依赖

xml

<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.0.24</version>
</dependency>
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.protocol</artifactId>
    <version>1.0.24</version>
</dependency>
<dependency>
    <groupId>org.apache.solr</groupId>
    <artifactId>solr-solrj</artifactId>
    <version>4.10.3</version>
</dependency>

同步

java

import cn.itcast.pojo.Book;
import cn.itcast.util.BinlogValue;
import cn.itcast.util.CanalDataParser;
import cn.itcast.util.DateUtils;
import cn.itcast.util.InnerBinlogEntry;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.solr.client.solrj.SolrServer;
import org.apache.solr.client.solrj.impl.HttpSolrServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.text.ParseException;
import java.util.List;
import java.util.Map;

public class SyncDataBootStart {

    private static Logger logger = LoggerFactory.getLogger(SyncDataBootStart.class);

    public static void main(String[] args) throws Exception {

        String hostname = "192.168.142.155";
        Integer port = 11111;
        String destination = "example";

        //获取CanalServer 连接
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(hostname, port), destination, "", "");

        //连接CanalServer
        canalConnector.connect();

        //订阅Destination
        canalConnector.subscribe();

        //轮询拉取数据
        Integer batchSize = 5*1024;
        while (true){
            Message message = canalConnector.getWithoutAck(batchSize);

            long messageId = message.getId();
            int size = message.getEntries().size();

            if(messageId == -1 || size == 0){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else{

                //进行数据同步
                //1. 解析Message对象
                List<InnerBinlogEntry> innerBinlogEntries = CanalDataParser.convertToInnerBinlogEntry(message);


                //2. 将解析后的数据信息 同步到Solr的索引库中.
                syncDataToSolr(innerBinlogEntries);


            }

            //提交确认
            canalConnector.ack(messageId);

        }

    }

    private static void syncDataToSolr(List<InnerBinlogEntry> innerBinlogEntries) throws Exception {

        //获取solr的连接
        SolrServer solrServer = new HttpSolrServer("http://192.168.142.143:8080/solr");

        //遍历数据集合 , 根据数据集合中的数据信息, 来决定执行增加, 修改 , 删除操作 .
        if(innerBinlogEntries != null){
            for (InnerBinlogEntry innerBinlogEntry : innerBinlogEntries) {

                CanalEntry.EventType eventType = innerBinlogEntry.getEventType();

                //如果是Insert, update , 则需要同步数据到 solr 索引库
                if(eventType == CanalEntry.EventType.INSERT || eventType == CanalEntry.EventType.UPDATE){
                    List<Map<String, BinlogValue>> rows = innerBinlogEntry.getRows();
                    if(rows != null){
                        for (Map<String, BinlogValue> row : rows) {
                            BinlogValue id = row.get("id");
                            BinlogValue name = row.get("name");
                            BinlogValue author = row.get("author");
                            BinlogValue publishtime = row.get("publishtime");
                            BinlogValue price = row.get("price");
                            BinlogValue publishgroup = row.get("publishgroup");

                            Book book = new Book();
                            book.setId(Integer.parseInt(id.getValue()));
                            book.setName(name.getValue());
                            book.setAuthor(author.getValue());
                            book.setPrice(Double.parseDouble(price.getValue()));
                            book.setPublishgroup(publishgroup.getValue());
                            book.setPublishtime(DateUtils.parseDate(publishtime.getValue()));


                            //导入数据到solr索引库
                            solrServer.addBean(book);
                            solrServer.commit();
                        }
                    }

                }else if(eventType == CanalEntry.EventType.DELETE){
                    //如果是Delete操作, 则需要删除solr索引库中的数据 .
                    List<Map<String, BinlogValue>> rows = innerBinlogEntry.getRows();
                    if(rows != null){
                        for (Map<String, BinlogValue> row : rows) {
                            BinlogValue id = row.get("id");

                            //根据ID删除solr的索引库
                            solrServer.deleteById(id.getValue());
                            solrServer.commit();

                        }
                    }
                }
            }
        }
    }

}

CanalDataParser 转换解析

用来转换解析从CanalServer中获取的Message对象.

java

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.SystemUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin;
import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd;
import com.google.protobuf.InvalidProtocolBufferException;

/**
 * 
 * ClassName: CanalDataParser <br/> 
 * 
 * @author Deng
 */
public class CanalDataParser {
	
	protected static final String DATE_FORMAT 	= "yyyy-MM-dd HH:mm:ss";
	protected static final String yyyyMMddHHmmss = "yyyyMMddHHmmss";
	protected static final String yyyyMMdd 		= "yyyyMMdd";
	protected static final String SEP 			= SystemUtils.LINE_SEPARATOR;
	protected static String  context_format     = null;
    protected static String  row_format         = null;
    protected static String  transaction_format = null;
    protected static String row_log = null;
	
	private static Logger logger = LoggerFactory.getLogger(CanalDataParser.class);
	
	static {
        context_format = SEP + "****************************************************" + SEP;
        context_format += "* Batch Id: [{}] ,count : [{}] , memsize : [{}] , Time : {}" + SEP;
        context_format += "* Start : [{}] " + SEP;
        context_format += "* End : [{}] " + SEP;
        context_format += "****************************************************" + SEP;

        row_format = SEP
                     + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms"
                     + SEP;

        transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {} , delay : {}ms" + SEP;

        row_log = "schema[{}], table[{}]";
    }



	public static List<InnerBinlogEntry> convertToInnerBinlogEntry(Message message) {
		List<InnerBinlogEntry> innerBinlogEntryList = new ArrayList<InnerBinlogEntry>();
		
		if(message == null) {
			logger.info("接收到空的 message; 忽略");
			return innerBinlogEntryList;
		}
		
		long batchId = message.getId();
        int size = message.getEntries().size();
        if (batchId == -1 || size == 0) {
        	logger.info("接收到空的message[size=" + size + "]; 忽略");
        	return innerBinlogEntryList;
        }

        printLog(message, batchId, size);
        List<Entry> entrys = message.getEntries();

        //输出日志
        for (Entry entry : entrys) {
        	long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;
        	
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) {
                    TransactionBegin begin = null;
                    try {
                        begin = TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    logger.info("BEGIN ----> Thread id: {}",  begin.getThreadId());
                    logger.info(transaction_format, new Object[] {entry.getHeader().getLogfileName(),
                                String.valueOf(entry.getHeader().getLogfileOffset()), String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime) });

                } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) {
                    TransactionEnd end = null;
                    try {
                        end = TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务提交信息，事务id
                    logger.info("END ----> transaction id: {}", end.getTransactionId());
                    logger.info(transaction_format,
                        new Object[] {entry.getHeader().getLogfileName(),  String.valueOf(entry.getHeader().getLogfileOffset()),
                                String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime) });
                }
                continue;
            }


            //解析结果
            if (entry.getEntryType() == EntryType.ROWDATA) {
                RowChange rowChage = null;
                try {
                    rowChage = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                EventType eventType = rowChage.getEventType();

                logger.info(row_format, new Object[] { entry.getHeader().getLogfileName(),
                            String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                            entry.getHeader().getTableName(), eventType, String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime) });


                //组装数据结果
                if (eventType == EventType.INSERT || eventType == EventType.DELETE || eventType == EventType.UPDATE) {
                	String schemaName = entry.getHeader().getSchemaName();
                	String tableName = entry.getHeader().getTableName();
                	List<Map<String, BinlogValue>> rows = parseEntry(entry);

                	InnerBinlogEntry innerBinlogEntry = new InnerBinlogEntry();
                	innerBinlogEntry.setEntry(entry);
                	innerBinlogEntry.setEventType(eventType);
                	innerBinlogEntry.setSchemaName(schemaName);
                	innerBinlogEntry.setTableName(tableName.toLowerCase());
                	innerBinlogEntry.setRows(rows);

                	innerBinlogEntryList.add(innerBinlogEntry);
                } else {
                	logger.info(" 存在 INSERT INSERT UPDATE 操作之外的SQL [" + eventType.toString() + "]");
                }
                continue;
            }
        }
		
		return innerBinlogEntryList;
	}





	private static List<Map<String, BinlogValue>> parseEntry(Entry entry) {
		List<Map<String, BinlogValue>> rows = new ArrayList<Map<String, BinlogValue>>();
		try {
			String schemaName = entry.getHeader().getSchemaName();
        	String tableName = entry.getHeader().getTableName();
			RowChange rowChage = RowChange.parseFrom(entry.getStoreValue());
			EventType eventType = rowChage.getEventType();

			// 处理每个Entry中的每行数据
			for (RowData rowData : rowChage.getRowDatasList()) {
				StringBuilder rowlog = new StringBuilder("rowlog schema[" + schemaName + "], table[" + tableName + "], event[" + eventType.toString() + "]");
				
				Map<String, BinlogValue> row = new HashMap<String, BinlogValue>();
				List<Column> beforeColumns = rowData.getBeforeColumnsList();
				List<Column> afterColumns = rowData.getAfterColumnsList();
				beforeColumns = rowData.getBeforeColumnsList();
			    if (eventType == EventType.DELETE) {//delete
			    	for(Column column : beforeColumns) {
			    		BinlogValue binlogValue = new BinlogValue();
			    		binlogValue.setValue(column.getValue());
			    		binlogValue.setBeforeValue(column.getValue());
				    	row.put(column.getName(), binlogValue);
				    }
			    } else if(eventType == EventType.UPDATE) {//update
			    	for(Column column : beforeColumns) {
			    		BinlogValue binlogValue = new BinlogValue();
			    		binlogValue.setBeforeValue(column.getValue());
				    	row.put(column.getName(), binlogValue);
				    }
			    	for(Column column : afterColumns) {
			    		BinlogValue binlogValue = row.get(column.getName());
			    		if(binlogValue == null) {
			    			binlogValue = new BinlogValue();
			    		}
			    		binlogValue.setValue(column.getValue());
				    	row.put(column.getName(), binlogValue);
				    }
			    } else { // insert
			    	for(Column column : afterColumns) {
			    		BinlogValue binlogValue = new BinlogValue();
			    		binlogValue.setValue(column.getValue());
			    		binlogValue.setBeforeValue(column.getValue());
				    	row.put(column.getName(), binlogValue);
				    }
			    } 
			    
			    rows.add(row);
			    String rowjson = JacksonUtil.obj2str(row);
			    
			    logger.info("########################### Data Parse Result ###########################");
			    logger.info(rowlog + " , " + rowjson);
			    logger.info("########################### Data Parse Result ###########################");
			    logger.info("");
			}
		} catch (InvalidProtocolBufferException e) {
			throw new RuntimeException("parseEntry has an error , data:" + entry.toString(), e);
		}
        return rows;
	}





	private static void printLog(Message message, long batchId, int size) {
        long memsize = 0;
        for (Entry entry : message.getEntries()) {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries())) {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        logger.info(context_format, new Object[] {batchId, size, memsize, format.format(new Date()), startPosition, endPosition });
    }



	private static String buildPositionForDump(Entry entry) {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":" + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
    }
	
}

InnerBinlogEntry 数据对象

用于封装解析后的数据对象 , 包含操作的是哪个数据库,那张表,操作类型,及本次操作的结果集.

java

public class InnerBinlogEntry {
   /**
    * canal原生的Entry
    */
   private Entry entry;
   /**
    * 该Entry归属于的表名
    */
   private String tableName;
   /**
    * 该Entry归属数据库名
    */
   private String schemaName;
   /**
    * 该Entry本次的操作类型，对应canal原生的枚举；EventType.INSERT; EventType.UPDATE; EventType.DELETE;
    */
   private EventType eventType;
   private List<Map<String, BinlogValue>> rows = new ArrayList<Map<String, BinlogValue>>();
  
   public Entry getEntry() {
      return entry;
   }
   public void setEntry(Entry entry) {
      this.entry = entry;
   }
   public String getTableName() {
      return tableName;
   }
   public void setTableName(String tableName) {
      this.tableName = tableName;
   }
   public EventType getEventType() {
      return eventType;
   }
   public void setEventType(EventType eventType) {
      this.eventType = eventType;
   }
   public String getSchemaName() {
      return schemaName;
   }
   public void setSchemaName(String schemaName) {
      this.schemaName = schemaName;
   }
   public List<Map<String, BinlogValue>> getRows() {
      return rows;
   }
   public void setRows(List<Map<String, BinlogValue>> rows) {
      this.rows = rows;
   }
}

BinlogValue 值

binlog分析的每行每列的value值

java

public class BinlogValue implements Serializable {

   private static final long serialVersionUID = -6350345408773943086L;
   
   private String value;
   private String beforeValue;
   /**
    * binlog分析的每行每列的value值；<br>
    * 新增数据： value：为现有值；<br>
    * 修改数据：value为修改后的值；<br>
    * 删除数据：value是删除前的值； 这个比较特殊主要是为了删除数据时方便获取删除前的值<br>
    */
   public String getValue() {
      return value;
   }
   public void setValue(String value) {
      this.value = value;
   }
   /**
    * binlog分析的每行每列的beforeValue值；<br>
    * 新增数据：beforeValue为现有值；<br>
    * 修改数据：beforeValue是修改前的值；<br>
    * 删除数据：beforeValue为删除前的值； <br>
    *
    */
   public String getBeforeValue() {
      return beforeValue;
   }
   public void setBeforeValue(String beforeValue) {
      this.beforeValue = beforeValue;
   }
}

三、安装

3.1 开启Mysql主从

Canal是基于Mysql的主从同步实现的，因此需要先开启Mysql的主从同步。

修改mysql配置文件

properties

# 设置binary log文件的存放地址和文件名，叫做mysql-bin


<NolebasePageProperties />




log-bin=/var/lib/mysql/mysql-bin
# 指定对哪个database记录binary log events
binlog-do-db=demo
#选择 row 模式
binlog_format=ROW
#配置mysql replaction需要定义，不能与canal的slaveId重复
server_id=1

binlog_format 配置

1)Row
	日志中会记录成每一行数据被修改的形式，然后在 slave 端再对相同的数据进行修改。
	优点：在 row 模式下，bin-log 中可以不记录执行的 SQL 语句的上下文相关的信息，仅仅只需要记录那一条记录被修改了，修改成什么样了。所以 row 的日志内容会非常清楚的记录下每一行数据修改的细节，非常容易理解。而且不会出现某些特定情况下的存储过程或 function ，以及 trigger 的调用和触发无法被正确复制的问题。
2)Statement
每一条会修改数据的 SQL 都会记录到 master 的 bin-log 中。slave 在复制的时候 SQL 进程会解析成和原来 master 端执行过的相同的 SQL 再次执行。
优点：在 statement 模式下，首先就是解决了 row 模式的缺点，不需要记录每一行数据的变化，减少了 bin-log 日志量，节省 I/O 以及存储资源，提高性能。因为他只需要记录在 master 上所执行的语句的细节，以及执行语句时候的上下文的信息。
缺点：在 statement 模式下，由于他是记录的执行语句，所以，为了让这些语句在 slave 端也能正确执行，那么他还必须记录每条语句在执行的时候的一些相关信息，也就是上下文信息，以保证所有语句在 slave 端杯执行的时候能够得到和在 master 端执行时候相同的结果。另外就是，由于 MySQL 现在发展比较快，很多的新功能不断的加入，使 MySQL 的复制遇到了不小的挑战，自然复制的时候涉及到越复杂的内容，bug 也就越容易出现。在 statement 中，目前已经发现的就有不少情况会造成 MySQL 的复制出现问题，主要是修改数据的时候使用了某些特定的函数或者功能的时候会出现，比如：sleep() 函数在有些版本中就不能被正确复制，在存储过程中使用了 last_insert_id() 函数，可能会使 slave 和 master 上得到不一致的 id 等等。由于 row 是基于每一行来记录的变化，所以不会出现类似的问题。

添加一个用于数据同步的账户

sql

create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
FLUSH PRIVILEGES;

重启mysql

shell

docker restart mysql
-- 查看状态
show master status;

3.2 安装Canal

添加网络并使mysql加入

shell

docker network create test

docker network connect test mysql

加载镜像并安装

shell

docker load -i canal.tar

docker run -p 11111:11111 --name canal \
-e canal.destinations=heima \
-e canal.instance.master.address=mysql:3306  \
-e canal.instance.dbUsername=canal  \
-e canal.instance.dbPassword=canal  \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false  \
-e canal.instance.filter.regex=heima\\..* \
--network heima \
-d canal/canal-server:v1.1.5

说明:

-p 11111:11111：这是canal的默认监听端口
canal.instance.mysql.slaveId : mysql集群配置中的serverId概念，需要保证和当前mysql集群中id唯一;
-e canal.instance.master.address=mysql:3306：数据库地址和端口，如果不知道mysql容器地址，可以通过docker inspect 容器id来查看
-e canal.instance.dbUsername=canal：数据库用户名
-e canal.instance.dbPassword=canal ：数据库密码
canal.instance.defaultDatabaseName : mysql链接时默认数据库;
canal.instance.connectionCharset : mysql 数据解析编码;
-e canal.instance.filter.regex=：要监听的表名称

表名称监听支持的语法：

mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\) 
常见例子：
1.  所有表：.*   or  .*\\..*
2.  canal schema下所有表： canal\\..*
3.  canal下的以canal打头的表：canal\\.canal.*
4.  canal schema下的一张表：canal.test1
5.  多个规则组合使用然后以逗号隔开：canal\\..*,mysql.test1,mysql.test2

一、概述 ​

1.1 配置文件 ​

canal.properties (系统根配置文件) ​

instance.properties (instance级别的配置文件，每个instance一份) ​

二、使用Demo ​

2.1 与springboot整合 ​

依赖 ​

配置 ​

监听器类 ​

2.2 基础应用 ​

依赖 ​

同步 ​

CanalDataParser 转换解析 ​

InnerBinlogEntry 数据对象 ​

BinlogValue 值 ​

三、安装 ​

3.1 开启Mysql主从 ​

修改mysql配置文件 ​

binlog_format 配置 ​

添加一个用于数据同步的账户 ​

重启mysql ​

3.2 安装Canal ​

添加网络并使mysql加入 ​

加载镜像并安装 ​

一、概述

1.1 配置文件

canal.properties (系统根配置文件)

instance.properties (instance级别的配置文件，每个instance一份)

二、使用Demo

2.1 与springboot整合

依赖

配置

监听器类

2.2 基础应用

依赖

同步

CanalDataParser 转换解析

InnerBinlogEntry 数据对象

BinlogValue 值

三、安装

3.1 开启Mysql主从

修改mysql配置文件

binlog_format 配置

添加一个用于数据同步的账户

重启mysql

3.2 安装Canal

添加网络并使mysql加入

加载镜像并安装